package com.atguigu.gamll.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.text.SimpleDateFormat;
import java.util.Properties;

/**
 * 日志数据分流
 * 需要启动的进程：zk、kafka、hdfs、logger(Nginx、日志采集服务器)、BaseLogApp
 *
 * 执行流程
 *      运行模拟生成日志数据的jar
 *      将生成的数据发送给Nginx进行负载均衡
 *      Nginx将日志数据转发给三台日志采集服务器进行处理
 *      日志采集服务器接收到日志之后
 *          打印
 *          落盘
 *          发送到kafka的ods_base_log主题
 *      当前服务打印ods_base_log主题中的数据
 */
public class BaseLogApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // TODO 2.检查点相关的设置
        BaseLogApp baseLogApp = new BaseLogApp();
        baseLogApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.从kafka中读取数据
        // 3.1 定义消费的主题以及消费者组
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        // 3.2 获取消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);
        // TODO 4.对读取的数据进行类型转换  jsonStr-->jsonObj
        // 4.1 匿名内部类
        /*
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );*/
        // 4.2 lambda表达式
        /*
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(
            jsonStr -> JSON.parseObject(jsonStr)
        );*/
        // 4.3 方法的默认调用
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        // TODO 5.对新老访客标记进行修复
        // 5.1 按照mid进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        // 5.2 使用flink的状态 判断是否曾经访问过，多标记进行修复
        //   修复思路：将当前设备第一次访问日期放到状态中存储起来，下次再进行访问的时候，先从状态中获取访问日期，
        //      如果获取到了，说明曾经访问过，将标记修复为0；如果从状态中没有获取到数据，说明是第一次访问，将本次访问日志放到状态中
        SingleOutputStreamOperator<JSONObject> jsonObjWithIsNewDS = keyedDS.map(
                new RichMapFunction<JSONObject, JSONObject>() {
                    // 注意：不能在声明状态的时候对其进行初始化，因为获取不到 RuntimeContext
                    private ValueState<String> lastVisitDateState;
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd");
                        lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDateState", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObj) throws Exception {
                        // 获取当前日志中新老访客标记
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        // 如果新老访客标记是1的话，有可能需要修复
                        if ("1".equals(isNew)) {
                            // 从状态中获取上次的访问日期
                            String lastVisitDate = lastVisitDateState.value();
                            // 当前访问日期
                            String curVisitDate = sdf.format(jsonObj.getLong("ts"));
                            // 判断是否曾经访问过
                            if (lastVisitDate != null && lastVisitDate.length() > 0) {
                                // 曾经访问过  对标记进行修复
                                isNew = "0";
                                jsonObj.getJSONObject("common").put("is_new", isNew);
                            } else {
                                // 第一次访问  将这次访问的日期放到状态中
                                lastVisitDateState.update(curVisitDate);
                            }
                        }
                        return jsonObj;
                    }
                }
        );
        // TODO 6.对日志数据进行分流  启动--启动侧输出流  曝光--曝光侧输出流  页面--主流
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag"){};

        SingleOutputStreamOperator<String> pageDS = jsonObjWithIsNewDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context ctx, Collector<String> out) throws Exception {
                        JSONObject startJsonObj = jsonObject.getJSONObject("start");
                        // 判断是否为启动日志
                        if (startJsonObj != null && startJsonObj.size() > 0) {
                            // 启动日志 -- 放到启动侧输出流
                            ctx.output(startTag, jsonObject.toJSONString());
                        } else {
                            // 除了启动日志之外，剩下的都属于页面日志 -- 放到主流中
                            out.collect(jsonObject.toJSONString());
                            // 判断在页面上是否有曝光信息
                            JSONArray displaysArr = jsonObject.getJSONArray("displays");
                            if (displaysArr != null && displaysArr.size() > 0) {
                                // 如果有曝光信息，对曝光数据进行遍历，将遍历出来的曝光数据 -- 放到曝光侧输出流中
                                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                                Long ts = jsonObject.getLong("ts");
                                for (int i = 0; i < displaysArr.size(); i++) {
                                    JSONObject displaysJsonObj = displaysArr.getJSONObject(i);
                                    // 放入页面id和时间戳
                                    displaysJsonObj.put("page_id", pageId);
                                    displaysJsonObj.put("ts", ts);
                                    ctx.output(displayTag, displaysJsonObj.toJSONString());
                                }
                            }
                        }
                    }
                }
        );
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);

        pageDS.print(">>>");
        startDS.print("启动>");
        displayDS.print("曝光>");

        // TODO 7.将不同流的数据写到Kafka的不同的主题中
        pageDS.addSink(MyKafkaUtil.getKafkaSink("dwd_page_log"));
        startDS.addSink(MyKafkaUtil.getKafkaSink("dwd_start_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaSink("dwd_display_log"));
    }
}
